package com.htby.bksw.imcmqtt.mqtt2;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.context.annotation.Configuration;

import java.util.LinkedList;

/**
 *  MqttClient 连接池
 * @author 周西栋
 * @date
 * @param
 * @return
 */
@Slf4j
@Configuration
public class MqttClientPool {


    private static MqttClientPool MQTTCLIENTPOOL = new MqttClientPool();

    private MqttClientPool(){}

    public static MqttClientPool instance(){
        return MQTTCLIENTPOOL;
    }

    // 连接池的最大容量
//    @Value("${mqttclientpool.maximal}")
    private static int maximal = 50;

    // 连接池的最小容量
//    @Value("${mqttclientpool.minimum}")
    private static int minimum = 20;

    // 连接池的初始容量
//    @Value("${mqttclientpool.initial}")
    private static int initial = 20;

    // 连接池的扩充容量
//    @Value("${mqttclientpool.step}")
    private static int step = 5;

    // 空闲连接池
    private static LinkedList<MqttClient> pool;

    // 已用连接池
    private static LinkedList<MqttClient> pool_used;

    // 静态代码块
    static {
        pool = new LinkedList<>();
        pool_used = new LinkedList<>();
        AutoMqttProperties properties = new AutoMqttProperties();
        MqttClientFactory factory = new MqttClientFactory();
        // 初始化连接池
        for (int i=0; i<minimum;i++){
            pool.add(factory.createMqttClient(properties));
        }
    }

    // 扩容
    private boolean dilatation(){
        if (pool.size() >= maximal){
            return false;
        }
        AutoMqttProperties properties = new AutoMqttProperties();
        MqttClientFactory factory = new MqttClientFactory();
        int limitnum = (pool.size() + step > maximal) ? (pool.size() + step - maximal) : step;
        // 扩容连接池
        for (int i=0; i<limitnum;i++){
            pool.add(factory.createMqttClient(properties));
        }
        log.info("连接池中现在还有{}个连接",pool.size());
        return true;
    }

    // 对外提供连接
    public MqttClient getClient(){
        MqttClient client = null;
        if(pool.size() > minimum || pool.size() > 0){
            client = pool.get(0);
            pool.remove(0);
            pool_used.add(client);
        } else if (pool.size() <= minimum || pool.size() == 0){
            dilatation();
            client = pool.get(0);
            pool.remove(0);
            pool_used.add(client);
        } else {
            log.error("实在抱歉，我是mqtt连接池，我已经没有连接可以提供了！");
            throw new RuntimeException("实在抱歉，连接池里已经没有连接了！");
        }
        return client;
    }

    // 归还连接
    public boolean close(MqttClient client){
        if (client == null){
            return false;
        }
        boolean b = pool_used.remove(client);
        if (b){
            if (pool.add(client)){
                return true;
            }else {
                pool_used.add(client);
                return false;
            }
        }else {
            return false;
        }
    }
}
